import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow

import scala.collection.mutable

/**
 * 滚动窗口（TumblingEventTimeWindows）
 * 结果是按照 Event Time 的时间窗口计算得出的，而无关系统的时间（包括输入的快慢）。
 */
object TumblingEventTimeWindowsTest {
  def main(args: Array[String]): Unit = {

    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    environment.setParallelism(1)

    val stream = environment.socketTextStream("localhost", 7777)
    val textKeyStream: KeyedStream[(String, Long, Int), Tuple] = stream.map {
      text => {
        val strings = text.split(",")
        (strings(0), strings(1).toLong, 1)
      }
    }.assignTimestampsAndWatermarks(
      new BoundedOutOfOrdernessTimestampExtractor[(String, Long, Int)](Time.milliseconds(1000)) {
        override def extractTimestamp(element: (String, Long, Int)): Long = {
          element._2
        }
      }
    ).keyBy(_._1)
    textKeyStream.print("textKeyBy:")

    val windowStream: WindowedStream[(String, Long, Int), Tuple, TimeWindow] = textKeyStream.window(TumblingEventTimeWindows.of(Time.seconds(2)))

    val groupDstream: DataStream[mutable.HashSet[Long]] =
      windowStream.fold(new mutable.HashSet[Long]()) { case (set, (key, ts, count))
      =>
        set += ts
      }
    groupDstream.print("window::::").setParallelism(1)

    environment.execute()
  }

}
